package SparkStreamingKafka;

import org.apache.commons.codec.StringDecoder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import scala.Tuple2;

import java.util.*;

/**
 * 基于Kafka Direct方式的实时wordCount程序
 */
public class KafkaDirectWordCount {


    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("KafkaDirectWordCount")
                .setMaster("local[2]");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        //1：创建一份Kafka参数map
        Map<String, String> kafkaParams = new HashMap<String,String>();
        //通过bin/kafka-console-producer.sh --broker-list获取
        kafkaParams.put("metadata.broker.list", "xxx");

        //2：创建set，里面放入要读取的topic
        Set<String> topics = new HashSet<String>();
        topics.add("WordCount");

       /* //3：输入DStream
        JavaPairInputDStream<String, String> lines = KafkaUtils.createDirectStream(
                jssc,
                String.class,
                String.class,
                StringDecoder.class,
                StringDecoder.class,
                kafkaParams,
                topics
        );*/

        //4:执行wordCount操作
      /*  JavaDStream<String> words = lines.flatMap(
                (FlatMapFunction<Tuple2<String, String>, String>) tuple2 -> Arrays.asList(tuple2._2.split(" "))
        );

        JavaPairDStream<String, Integer> pairs = words.mapToPair(
                (PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1)
        );

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(
                (Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2
        );

        wordCounts.print();*/

        jssc.start();
        //jssc.awaitTermination();
        jssc.close();
    }
}
